package com.shujia.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo2UpdateStateByKey {
  def main(args: Array[String]): Unit = {

    /**
      * 创建spark 环境
      *
      */

    val conf: SparkConf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("wc")


    val sc = new SparkContext(conf)


    /**
      * 创建spark streaming环境
      * 指定每隔多久计算一次
      *
      */

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    //设置checkpoint路径
    ssc.checkpoint("Spark/data/checkpoint")


    /**
      * 读取数据
      *
      */

    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    val wordsDS: DStream[String] = linesDS.flatMap(_.split(","))

    val kvDS: DStream[(String, Int)] = wordsDS.map((_, 1))

    /**
      * 有状态算子
      *
      * 需要设置checkpoint 用于保存之前的计算状态
      *
      */


    /**
      *
      * 更新函数： 使用当前batch的数据去更新之前的计算结果，返回一个新的结果
      *
      * seq: 当前batch一个单词所有的value
      * opt: 之前一个单词的计算结果（状态）,  使用Option的原因是如果是第一次计算之前没有结果，那就是NOne
      */
    val updateFun = (seq: Seq[Int], opt: Option[Int]) => {

      //计算当前batch单词的数量
      val currCount: Int = seq.sum

      //获取之前的计算结果
      val beCount: Int = opt.getOrElse(0)

      //最新一个单词的数量
      val newCount: Int = currCount + beCount

      //返回Option
      Option(newCount)
    }


    val countDS: DStream[(String, Int)] = kvDS.updateStateByKey(updateFun)

    countDS.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()




  }

}
